//------------------------------------------------------------------------------
// <copyright file="RpcChannelListener.cpp" company="Microsoft">
//     Copyright (c) Microsoft Corporation.  All rights reserved.
// </copyright>
//------------------------------------------------------------------------------

#include "stdafx.h"

#include "RpcChannelListener.h"
#include "RpcSocketChannel.h"

#ifdef _DURANGO
	// For Durango platform, use Winsock
	#include "WinsockListeningSocket.h"
#else
	#if WINAPI_FAMILY_PARTITION(WINAPI_PARTITION_DESKTOP)
		// For Win32 platform, use Winsock
		#include "WinsockListeningSocket.h"
	#else
		// For WinRT platform, use WinRT sockets
		#include "WinRtSocketListener.h"
	#endif
#endif

using namespace Microsoft::Internal::GamesTest::Rpc::Server;

namespace
{
	static std::mutex s_instanceLock;
	const RpcChannelListener::IncomingMessageCallback EmptyIncomingMessageCallback = [](std::weak_ptr<IRpcChannel>, RpcSocketTransportMessage){};
}

RpcChannelListener* RpcChannelListener::s_instance = nullptr;

RpcChannelListener::RpcChannelListener()
	: m_isStarted(false), m_incomingMessageCallback(EmptyIncomingMessageCallback)
{
}

RpcChannelListener* RpcChannelListener::Instance()
{
	// It's not intended for this function to be called very often so it's OK to acquire
	// the lock on every call in order to avoid all of the multi-threading issues that can
	// arise with the Singleton pattern.
	std::unique_lock<std::mutex> lock(s_instanceLock);

	if(s_instance == nullptr)
	{
#ifdef _DURANGO
		// For Durango platform, use Winsock
		s_instance = new WinsockListeningSocket();
#else
	#if WINAPI_FAMILY_PARTITION(WINAPI_PARTITION_DESKTOP)
		// For Win32 platform, use Winsock
		s_instance = new WinsockListeningSocket();
	#else
		// For WinRT platform, use WinRT sockets
		s_instance = new WinRtSocketListener();
	#endif
#endif
		
	}

	return s_instance;
}

GAMESTEST_ANNOTATE_HRESULT_METHOD HRESULT RpcChannelListener::Start(int port, IncomingMessageCallback incomingMessageCallback, ClientDisconnectedCallback clientDisconnectedCallback)
{
	std::unique_lock<std::mutex> lock(s_synchronizationContext);

	HRESULT hr = E_FAIL;
	if(m_isStarted)
	{
		hr = GAMESTEST_RPC_S_ALREADY_LISTENING;
		goto end;
	}

	m_incomingMessageCallback = incomingMessageCallback;
	m_clientDisconnectedCallback = clientDisconnectedCallback;

	hr = this->StartImpl(port);

	if(SUCCEEDED(hr))
	{
		m_isStarted = true;
	}
	else
	{
		goto end;
	}

end:
	return hr;
}

GAMESTEST_ANNOTATE_HRESULT_METHOD HRESULT RpcChannelListener::Stop()
{
	std::unique_lock<std::mutex> lock(s_synchronizationContext);

	HRESULT hr = S_OK;

	if(!m_isStarted)
	{
		hr = GAMESTEST_RPC_S_NOT_STARTED;
		goto end;
	}

	hr = this->StopImpl();

	m_clients.clear();

	m_incomingMessageCallback = EmptyIncomingMessageCallback;

end:
	m_isStarted = false;

	return hr;
}

void RpcChannelListener::OnSocketConnected(IRpcStringSocket* socket)
{
	HRESULT hr;

	std::unique_lock<std::mutex> lock(s_synchronizationContext);

	if(m_isStarted)
	{
		std::shared_ptr<IRpcChannel> client = CreateRpcChannel(socket);
		IRpcChannel::RequestReceivedCallback requestReceived = std::bind(&RpcChannelListener::OnChannelRequestReceived, this, std::placeholders::_1, std::placeholders::_2);
		IRpcChannel::ChannelDisconnectedCallback channelDisconnected = std::bind(&RpcChannelListener::OnChannelDisconnected, this, std::placeholders::_1);
		hr = client->StartListening(requestReceived, channelDisconnected);
		if(FAILED(hr))
		{
			RpcPrintMessage(L"Failed to start listening for requests on newly connected client:");
			RpcPrintMessage(socket->GetIpAddress().c_str());
			RpcPrintError(hr);

			// This will destroy the shared pointer and close the socket.
			client = nullptr;
		}
		else
		{
			m_clients.push_back(client);
		}
	}
}

void RpcChannelListener::OnChannelRequestReceived(std::weak_ptr<IRpcChannel> client, RpcSocketTransportMessage requestMessage)
{
	m_incomingMessageCallback(client, requestMessage);
}

void RpcChannelListener::OnChannelDisconnected(std::weak_ptr<IRpcChannel> client)
{
	std::unique_lock<std::mutex> lock(s_synchronizationContext);

	std::shared_ptr<IRpcChannel> sharedClient = client.lock();
	if(sharedClient != nullptr)
	{
		ClientCollection::iterator iter = std::find_if(m_clients.begin(), m_clients.end(), [=](std::shared_ptr<IRpcChannel> input) { return input == sharedClient; } );
		if(iter != m_clients.end())
		{
			m_clients.erase(iter);
		}

		m_clientDisconnectedCallback(client);
	}
}

std::shared_ptr<IRpcChannel> RpcChannelListener::CreateRpcChannel(IRpcStringSocket* socket)
{
	return std::make_shared<RpcSocketChannel>(socket);
}